package com.sanq.product.books.kafka.producer.service.impl;

import com.sanq.product.books.entity.vo.BooksVo;
import com.sanq.product.books.kafka.producer.service.BooksMQService;
import com.sanq.product.config.utils.web.JsonUtil;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

/**
 * com.sanq.product.books.kafka.producer.service.BooksMQServiceImpl
 *
 * @author sanq.Yan
 * @date 2019/8/13
 */
@Service("booksMQService")
public class BooksMQServiceImpl implements BooksMQService {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    @Resource
    private ThreadPoolTaskExecutor taskExecutor;

    @Value("${kafka.books.del}")
    public String kafkaBooksDelTopic;

    @Value("${kafka.chapter.del}")
    public String kafkaChapterDelTopic;

    @Value("${kafka.books.search}")
    public String kafkaBooksSearch;


    @Override
    public void deleteByBooksId(Long booksId) {
        taskExecutor.execute(() -> {
            kafkaTemplate.send(kafkaBooksDelTopic, booksId + "");
        });
    }

    @Override
    public void deleteByChapterId(Long chapterId) {
        taskExecutor.execute(() -> {
            kafkaTemplate.send(kafkaChapterDelTopic, chapterId + "");
        });
    }

    @Override
    public void save2Es(Long booksId, String type) {
        taskExecutor.execute(() -> {
            kafkaTemplate.send(kafkaBooksSearch, type,booksId.toString());
        });
    }

    @Override
    public void save2Es(List<BooksVo> booksVos, String type) {
        taskExecutor.execute(() -> {
            kafkaTemplate.send(kafkaBooksSearch, type, JsonUtil.obj2Json(booksVos));
        });
    }
}
